/*
 * Copyright 2023 yifeng Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.yifeng.repo.flink.data.transport.streaming.connectors.oracle;

import static org.apache.flink.util.Preconditions.checkNotNull;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

import javax.annotation.Nullable;

import com.google.common.base.Strings;
import com.yifeng.repo.flink.data.transport.config.DataSourceConfig;
import com.yifeng.repo.flink.data.transport.config.DataSourceConfigFactory;
import com.yifeng.repo.flink.data.transport.dto.DbTypeEnums;
import com.yifeng.repo.flink.data.transport.streaming.functions.FullSourceFunction;

/**
 * A builder to build a SourceFunction which can read snapshot and continue to consume log miner.
 */
public class OracleSource {

    public static <T> Builder<T> builder() {
        return new Builder<>();
    }

    /** Builder class of {@link OracleSource}. */
    public static class Builder<T> {

        private Integer port = 1521; // default 1521 port
        private String hostname;
        private String database;
        private String username;
        private String password;
        private String url;
        private String fetchSize;
        private String tableList;
        private String schemaList;
        private Map<String,List<Object>> tableFieldMap;
        //全量读取sql语句
        private Map<String,String> sqlCondition;
        //多表并行的并行度
        private Integer parallelism;

        public Builder<T> hostname(@Nullable String hostname) {
            this.hostname = hostname;
            return this;
        }

        /** Integer port number of the Oracle database server. */
        public Builder<T> port(@Nullable Integer port) {
            this.port = port;
            return this;
        }

        /** Url to use when connecting to the Oracle database server. */
        public Builder<T> url(@Nullable String url) {
            this.url = url;
            return this;
        }

        /**
         * An optional list of regular expressions that match database names to be monitored; any
         * database name not included in the whitelist will be excluded from monitoring. By default
         * all databases will be monitored.
         */
        public Builder<T> database(String database) {
            this.database = database;
            return this;
        }

        /** Name of the Oracle database to use when connecting to the Oracle database server. */
        public Builder<T> username(String username) {
            this.username = username;
            return this;
        }

        /** Password to use when connecting to the Oracle database server. */
        public Builder<T> password(String password) {
            this.password = password;
            return this;
        }

        public Builder<T> fetchSize(String fetchSize) {
            this.fetchSize = fetchSize;
            return this;
        }
        
        public Builder<T> tableList(String tableList) {
            this.tableList = tableList;
            return this;
        }
        
        public Builder<T> schemaList(String schemaList) {
            this.schemaList = schemaList;
            return this;
        }
        
        public Builder<T> tableFieldMap(Map<String,List<Object>> tableFieldMap) {
            this.tableFieldMap = tableFieldMap;
            return this;
        }
        
        public Builder<T> sqlCondition(Map<String,String> sqlCondition) {
            this.sqlCondition = sqlCondition;
            return this;
        }
        
        public Builder<T> parallelism(Integer parallelism) {
            this.parallelism = parallelism;
            return this;
        }
        
        
        public FullSourceFunction<T> build() {
        	DataSourceConfig config = new DataSourceConfigFactory().create();
        	config.setDriverClassName("oracle.jdbc.OracleDriver");
        	config.setDbType(DbTypeEnums.ORACLE.getType());
        	config.setHostname(checkNotNull(hostname));
        	config.setUsername(checkNotNull(username));
        	config.setPassword(checkNotNull(password));
        	config.setPort(port);
        	config.setJdbcUrl(url);
        	config.setDatabase(database);
        	config.setTableList(Arrays.asList(checkNotNull(tableList).split(",")));
        	config.setSchemaList(Arrays.asList(checkNotNull(schemaList).split(",")));
        	config.setParallelism(parallelism);
        	if (tableFieldMap != null) {
        		config.setTableFieldMap(tableFieldMap);
            }
        	if (sqlCondition != null) {
        		config.setSqlCondition(sqlCondition);
            }
        	if(!Strings.isNullOrEmpty(fetchSize)) {
        		config.setFetchSize(Integer.valueOf(fetchSize));
        	}
            return new FullSourceFunction<>(config,new OracleValidator());
        }
    }
}
